package com.example.transformation;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: RichFunction
 * Package: com.example.transformation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-22
 * Time: 18:39
 */

//富函数
public class RichFunction {
    public static void main(String[] args) throws Exception {

        //创建本地执行环境 和 web UI
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> water = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {

                        final String[] s = value.split(" ");

                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                });

        SingleOutputStreamOperator<WaterSensor> richReduce = water.keyBy(value -> value.getTs())
                .reduce(new RichReduceFunction<WaterSensor>() {

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //没有指定线程数 一启动默认舒适化8次
                        System.out.println("reduce聚合初始化操作");
                        //获取运行时上下文
                        RuntimeContext runtimeContext = getRuntimeContext();
                    }

                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                        return new WaterSensor(value1.getId(), Long.valueOf(value1.getTs()), Integer.valueOf(value2.getVc()));
                    }

                    @Override
                    public void close() throws Exception {
                        System.out.println("reduce聚合结束关闭操作");
                    }
                });

        richReduce.print();


        env.execute();

    }
}
